package com.google.code.p.restrpc.spring;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.util.Assert;

import com.google.code.p.restrpc.core.RpcServiceMetadata;
import com.google.code.p.restrpc.core.RpcServiceRepository;
import com.google.code.p.restrpc.core.RpcServiceVisitor;
import com.google.code.p.restrpc.core.event.ServiceRegisterEventListener;
import com.google.code.p.restrpc.core.event.ServiceRegisteredEventObject;
import com.google.code.p.restrpc.core.event.ServiceUnregisteredEventObject;
import com.google.code.p.restrpc.core.registry.RegistryClient;
import com.google.code.p.restrpc.core.registry.RegistryClientFactory;
import com.google.code.p.restrpc.event.RpcEventObject;
import com.google.code.p.restrpc.message.DefaultMessageProcessor;
import com.google.code.p.restrpc.message.MessageAction;
import com.google.code.p.restrpc.message.MessageContext;
import com.google.code.p.restrpc.message.MessageSender;
import com.google.code.p.restrpc.spring.heartbeat.ExportHeartBeater;

/**
 * @author dz
 */
public class RpcServiceRemoteExporter implements
                                     ServiceNodeDefinition,
                                     ServiceRegisterEventListener,
                                     InitializingBean,
                                     DisposableBean,
                                     ApplicationListener {
    
    private static final Log logger = LogFactory.getLog(RpcServiceRemoteExporter.class);
    
    private static RpcServiceRemoteExporter instance;
    
    public static RpcServiceRemoteExporter getInstance() {
        return instance;
    }
    
    private DefaultMessageProcessor messageProcessor;
    
    private String registryUrl;
    
    private String protocal = "http";
    
    private String address;
    
    private int port = 80;
    
    private String context = "restrpc";
    
    private String servicePrefix;
    
    private int weight = 1;
    
    private int heartbeat = 60 * 1000;
    
    private ExportHeartBeater exportHeartBeater;
    
    private RegistryClient registryClient;
    
    /**
     * 
     */
    public RpcServiceRemoteExporter() {
    }
    
    /**
     * @return the protocal
     */
    public String getProtocal() {
        return protocal;
    }
    
    /**
     * @param protocal
     *            the protocal to set
     */
    public void setProtocal(String protocal) {
        this.protocal = protocal;
    }
    
    /**
     * @return the address of current node , could be domain name or IP address
     */
    public String getAddress() {
        return address;
    }
    
    /**
     * @param address
     */
    public void setAddress(String address) {
        this.address = address;
    }
    
    /**
     * @return the port of current node, default value is 80
     */
    public int getPort() {
        return port;
    }
    
    /**
     * @param port
     */
    public void setPort(int port) {
        if (port < 80) {
            throw new IllegalArgumentException("The 'port' must be >= 80!");
        }
        
        this.port = port;
    }
    
    /**
     * @return the web rpc context of rpc service , default value is restrpc
     */
    public String getContext() {
        return context;
    }
    
    /**
     * @param rpcContext
     */
    public void setContext(String rpcContext) {
        if (StringUtils.isEmpty(rpcContext)) {
            throw new IllegalArgumentException("The 'rpcContext' can't be null or empty!");
        }
        this.context = rpcContext;
    }
    
    /**
     * @return
     */
    public String getRegistryUrl() {
        return registryUrl;
    }
    
    /**
     * @param registryUrl
     */
    public void setRegistryUrl(String registryUrl) {
        if (StringUtils.isEmpty(registryUrl)) {
            throw new IllegalArgumentException("The registryUrl can't be null or empty!");
        }
        this.registryUrl = registryUrl;
    }
    
    /**
     * @return
     */
    public int getHeartbeat() {
        return heartbeat;
    }
    
    /**
     * @param heartbeat
     */
    public void setHeartbeat(int heartbeat) {
        if (heartbeat <= 0) {
            throw new IllegalArgumentException("The 'heartbeat' must be > 0");
        }
        this.heartbeat = heartbeat;
    }
    
    /**
     * @return
     */
    public int getWeight() {
        return weight;
    }
    
    /**
     * @param weight
     */
    public void setWeight(int weight) {
        if (weight < 1) {
            throw new IllegalArgumentException("The 'weight' must be >= 1");
        }
        this.weight = weight;
    }
    
    /**
     * @return
     */
    public RegistryClient getRegistryClient() {
        return this.registryClient;
    }
    
    /**
     * @return
     */
    public String getServicePrefix() {
        if (!StringUtils.isEmpty(this.servicePrefix)) {
            return servicePrefix;
        }
        
        initialzeServicePrefix();
        
        return servicePrefix;
    }
    
    private synchronized void initialzeServicePrefix() {
        if (!StringUtils.isEmpty(this.servicePrefix)) {
            return;
        }
        
        this.servicePrefix = buildServicePrefix();
    }
    
    private String buildServicePrefix() {
        StringBuilder sb = new StringBuilder(getProtocal());
        sb.append("://").append(getAddress());
        if (getPort() != 80) {
            sb.append(":").append(getPort());
        }
        sb.append("/").append(getContext()).append("/");
        
        return sb.toString();
    }
    
    // ************************ServiceRegisterEventListener*************************
    /**
     * 
     */
    public boolean support(RpcEventObject eventObject) {
        return (eventObject instanceof ServiceUnregisteredEventObject);
    }
    
    /**
     * 
     */
    public void handle(RpcEventObject eventObject) {
        messageProcessor.addMessage(eventObject);
    }
    
    // ************************InitializingBean*************************
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(protocal);
        Assert.notNull(address);
        Assert.notNull(registryUrl);
        if (!"http".equals(protocal) && !"https".equals(protocal)) {
            throw new IllegalArgumentException("'protocal':only 'http' or 'https' allowed!");
        }
        //
        RpcServiceRepository.getInstance().setWeight(weight);
        //
        messageProcessor = new DefaultMessageProcessor();
        //
        registryClient = RegistryClientFactory.getInstance(getRegistryUrl());
        //
        instance = this;
    }
    
    // ************************DisposableBean*************************
    public void destroy() throws Exception {
        RegistryClientFactory.getInstance(getRegistryUrl())
                             .unregisterAll(getAddress(), getPort());
        RpcServiceRepository.getInstance().unexport();
        if (exportHeartBeater != null) {
            try {
                exportHeartBeater.stop();
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    // ************************ApplicationListener*************************
    public void onApplicationEvent(ApplicationEvent event) {
        if (event == null) {
            return;
        }
        if (event instanceof ContextRefreshedEvent) {
            if (logger.isDebugEnabled()) {
                logger.debug("Start export service and register remotely by async");
            }
            
            messageProcessor.put(ServiceRegisteredEventObject.class,
                                 new MessageSender(new MessageAction() {
                                     
                                     public void execute(MessageContext context) {
                                         ServiceRegisteredEventObject message = (ServiceRegisteredEventObject) context.getMessage();
                                         RpcServiceMetadata rpcServiceMetadata = message.getServiceMetadata();
                                         registryClient.registerService(rpcServiceMetadata.getServiceName(),
                                                                        getServicePrefix() + rpcServiceMetadata.getServiceName(),
                                                                        weight);
                                         rpcServiceMetadata.export();
                                         if (!RpcServiceRepository.getInstance()
                                                                  .isExported()) {
                                             RpcServiceRepository.getInstance()
                                                                 .export();
                                         }
                                     }
                                 },
                                                   true));
            
            messageProcessor.put(ServiceUnregisteredEventObject.class,
                                 new MessageSender(new MessageAction() {
                                     
                                     public void execute(MessageContext context) {
                                         ServiceUnregisteredEventObject message = (ServiceUnregisteredEventObject) context.getMessage();
                                         RpcServiceMetadata rpcServiceMetadata = message.getRpcServiceMetadata();
                                         registryClient.unregisterService(rpcServiceMetadata.getServiceName(),
                                                                          getServicePrefix() + rpcServiceMetadata.getServiceName());
                                     }
                                 },
                                                   true));
            
            // TODO REFACTOR to independence Class
            // Prepare register and execute in ASYNC
            new Thread(new Runnable() {
                
                /**
                 * After the service node startup ,to detect the service node
                 * has been registered at first.
                 * <p/>
                 * why need this and how this (has been registered)
                 * happened?because the node is shutdown accidently and the
                 * destroy method (to clear-up and unregister in remote register
                 * directory) can't be executed as expect.
                 * <p/>
                 * when the service node re-startup , it's must to detect
                 * <p/>
                 * if registered, must unregister all to pretend to register
                 * twice and more.
                 */
                public void run() {
                    
                    // prepare register
                    if (registryClient.isRegistered(getAddress(), getPort())) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("Detected:" + getAddress()
                                         + ":"
                                         + getPort()
                                         + "is registered!");
                        }
                        registryClient.unregisterAll(getAddress(), getPort());
                        RpcServiceRepository.getInstance().unexport();
                    }
                    
                    // start the heart-beat worker over HTTP
                    
                    // do register
                    RpcServiceRepository.getInstance()
                                        .accept(new RpcServiceVisitor() {
                                            
                                            public void visit(RpcServiceMetadata metadata) {
                                                if (!metadata.isExported()) {
                                                    messageProcessor.addMessage(new ServiceRegisteredEventObject(metadata));
                                                }
                                            }
                                            
                                        });
                    
                    // start heartBeatExportMonitor
                    exportHeartBeater = new ExportHeartBeater(RpcServiceRemoteExporter.this,
                                                              getHeartbeat());
                    exportHeartBeater.start();
                }
                
            }).run();
            
        }
    }
    
}
